package com.lzq.mq.nameserver.event.handler;

import com.alibaba.fastjson2.JSON;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.dto.HeartBeatDTO;
import com.lzq.mq.common.dto.PullBrokerIpReqDTO;
import com.lzq.mq.common.dto.PullBrokerIpRespDTO;
import com.lzq.mq.common.dto.ServiceRegistryReqDTO;
import com.lzq.mq.common.enums.NameServerEventCode;
import com.lzq.mq.common.event.EventBus;
import com.lzq.mq.common.event.model.Event;
import com.lzq.mq.nameserver.event.model.HeartBeatEvent;
import com.lzq.mq.nameserver.event.model.PullBrokerIpEvent;
import com.lzq.mq.nameserver.event.model.RegistryEvent;
import com.lzq.mq.nameserver.event.model.UnRegistryEvent;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.internal.StringUtil;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;

@Slf4j
@ChannelHandler.Sharable   // 表示所有链接器实例都共享一个实例, 防止每次链接器实例都创建一个实例
public class TcpNettyServerHandler extends SimpleChannelInboundHandler<TcpMsg> {

    /**
     * 事件总线
     */
    private EventBus eventBus;

    public TcpNettyServerHandler(EventBus eventBus) {
        this.eventBus = eventBus;
        log.info("初始化TcpNettyServerHandler");
    }

    // 1.网络请求的接受
    // 2.事件发不起的实现 事件总线->event
    // 3.事件处理器的实现
    // 4.数据存储
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TcpMsg tcpMsg) throws Exception {

        int code = tcpMsg.getCode();
        byte[] body = tcpMsg.getBody();
        Event event = null;
        if (NameServerEventCode.REGISTRY.getCode() == code) {
            // todo 用同一个对象不行吗, 为什么非要多此一举!
            ServiceRegistryReqDTO serviceRegistryReqDTO = JSON.parseObject(body, ServiceRegistryReqDTO.class);
            RegistryEvent registryEvent = new RegistryEvent();
            registryEvent.setRegistryType(serviceRegistryReqDTO.getRegistryType());
            registryEvent.setMsgId(serviceRegistryReqDTO.getMsgId());
            registryEvent.setPassword(serviceRegistryReqDTO.getPassword());
            registryEvent.setUser(serviceRegistryReqDTO.getUser());
            registryEvent.setAttrs(serviceRegistryReqDTO.getAttrs());

            if (StringUtil.isNullOrEmpty(serviceRegistryReqDTO.getIp())) {
            // 直接使用链接器的地址
                InetSocketAddress inetSocketAddress = (InetSocketAddress) channelHandlerContext.channel().remoteAddress();
                registryEvent.setPort(inetSocketAddress.getPort());
                registryEvent.setIp(inetSocketAddress.getHostString());
            } else {
                registryEvent.setPort(serviceRegistryReqDTO.getPort());
                registryEvent.setIp(serviceRegistryReqDTO.getIp());
            }
            event = registryEvent;
        } else if (NameServerEventCode.UN_REGISTRY.getCode() == code) {
            event = new UnRegistryEvent();
        } else if (NameServerEventCode.HEART_BEAT.getCode() == code) {

            HeartBeatDTO heartBeatEvent = JSON.parseObject(body, HeartBeatDTO.class);
            event = new HeartBeatEvent();
            event.setMsgId(heartBeatEvent.getMsgId());
        } else if (NameServerEventCode.PULL_BROKER_IP_LIST.getCode() == code) {
            // 拉取broker节点ip请求
            PullBrokerIpReqDTO dto = JSON.parseObject(body, PullBrokerIpReqDTO.class);
            PullBrokerIpEvent pullBrokerIpEvent = new PullBrokerIpEvent();
            pullBrokerIpEvent.setMsgId(dto.getMsgId());
            pullBrokerIpEvent.setRole(dto.getRole());
            pullBrokerIpEvent.setGroup(dto.getBrokerClusterGroup());
            event = pullBrokerIpEvent;
        }

        if (event == null) {
            log.error("无法解析的事件{}", tcpMsg);
            return;
        }

        event.setChannelHandlerContext(channelHandlerContext);
        eventBus.publish(event);
    }

    /*
        ChannelHandlerContext 的生命周期与关联的 Channel 紧密相关
        一对一关系: 每个 ChannelHandler 只有一个对应的 ChannelHandlerContext
        不跨Handler共享: 不同的 ChannelHandler 拥有各自独立的 ChannelHandlerContext 实例
        属性可共享: 虽然 ChannelHandlerContext 不共享，但通过 channel().attr() 设置的属性可在同一 Channel 的所有处理器间共享
     */

    /**
     * 异常处理
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }

    /**
     * 断开连接处理
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);

        // 如果依赖任务剔除节点, 会有三个心跳周期的延迟
        // 想要做到链接断开, 立即剔除, 可以通过这个回调函数发送UnRegister时间
        log.info("客户端断开连接:{}", ctx.channel().remoteAddress());
        UnRegistryEvent unRegistryEvent = new UnRegistryEvent();
        unRegistryEvent.setChannelHandlerContext(ctx);
        eventBus.publish(unRegistryEvent);
    }
}
